#pragma once

#include "requestor.hpp"
#include "../common/uuid.hpp"

namespace JsonMuduoRPC
{
    namespace client
    {
        class TopicManager
        {
        public:
            using ptr = std::shared_ptr<TopicManager>;
            using SubscriveCallBack = std::function<void(const std::string&, const std::string&)>;

            TopicManager(Requestor::ptr requestor) : _requestor(requestor) {}

            bool create(const BaseConnection::ptr &conn, const std::string &topic_name)
            {
                return commonRequest(conn, topic_name, TopicOpType::TOPIC_CREATE);
            }

            bool remove(const BaseConnection::ptr &conn, const std::string &topic_name)
            {
                return commonRequest(conn, topic_name, TopicOpType::TOPIC_REMOVE);
            }

            bool subscribe(const BaseConnection::ptr &conn, const std::string &topic_name, const SubscriveCallBack &cb)
            {
                addSubscribe(topic_name, cb);
                bool ret = commonRequest(conn, topic_name, TopicOpType::TOPIC_SUBSCRIBE);
                if (ret == false)
                {
                    delSubsribe(topic_name);
                    return false;
                }
                return true;
            }

            bool cancel(const BaseConnection::ptr &conn, const std::string &topic_name)
            {
                delSubsribe(topic_name);
                return commonRequest(conn, topic_name, TopicOpType::TOPIC_CANCEL);
            }

            bool publish(const BaseConnection::ptr &conn, const std::string &topic_name, const std::string &msg)
            {
                return commonRequest(conn, topic_name, TopicOpType::TOPIC_PUBLISH, msg);
            }

            // 处理服务端消息发布的请求——收到不同主题的消息，调用对应的回调函数进行处理
            // 交给底层的dispatcher模块
            void onPublish(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                if (msg->getOpType() != TopicOpType::TOPIC_PUBLISH)
                {
                    LOG(ERROR, "收到了错误的请求");
                    return;
                }

                SubscriveCallBack cb = getSbuscribe(msg->getTopic());
                if (cb == nullptr)
                {
                    LOG(ERROR, "没有主题 %s 对应的回调函数", msg->getTopic());
                    return;
                }

                cb(msg->getTopic(), msg->getMessage());
            }

        private:
            void addSubscribe(const std::string& topic_name, const SubscriveCallBack& cb)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _subscribe_cb_map.insert(std::make_pair(topic_name, cb));
            }

            void delSubsribe(const std::string& topic_name)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _subscribe_cb_map.erase(topic_name);
            }

            const SubscriveCallBack getSbuscribe(const std::string& topic_name)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _subscribe_cb_map.find(topic_name);
                if (it == _subscribe_cb_map.end())
                    return SubscriveCallBack();
                
                return it->second;

            }

            bool commonRequest(const BaseConnection::ptr &conn, const std::string &topic_name,
                               const TopicOpType opType, const std::string &msg = "")
            {
                TopicRequest::ptr topic_req = MessageFactory::create<TopicRequest>();
                topic_req->setId(UUID::getUuidStr());
                topic_req->setMType(MType::REQ_TOPIC);
                topic_req->setOpType(opType);
                topic_req->setTopic(topic_name);
                if (opType == TopicOpType::TOPIC_PUBLISH)
                    topic_req->setMessage(msg);

                BaseMessage::ptr msg_rsp;
                bool ret = _requestor->send(conn, topic_req, msg_rsp);
                if (ret == false)
                {
                    LOG(ERROR, "请求发送失败");
                    return false;
                }

                TopicResponse::ptr topic_rsp = std::dynamic_pointer_cast<TopicResponse>(msg_rsp);
                if (topic_rsp.get() == nullptr)
                {
                    LOG(ERROR, "响应类型转换失败");
                    return false;
                }
                if (topic_rsp->getRCode() != RCode::RCODE_OK)
                {
                    LOG(ERROR, "请求出错: %s", errReason(topic_rsp->getRCode()));
                    return false;
                }
                return true;
            }

        private:
            std::mutex _mutex;
            Requestor::ptr _requestor;
            std::unordered_map<std::string, SubscriveCallBack> _subscribe_cb_map; // 不同的主题有不同的回调
        };
    } // namespace client

}